package david.java.flink.function.custom;

import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple2;

/**
 * @Description:
 * @Author: David
 * @Date: Create in 下午12:10 2021/1/18
 */
public class MyAverageAggregateForProcess implements AggregateFunction<Tuple2<String ,Long>, Tuple2<Long, Long>, Double> {
    @Override
    public Tuple2<Long, Long> createAccumulator() {
        return new Tuple2<>(0L, 0L);
    }

    @Override
    public Tuple2<Long, Long> add(Tuple2<String, Long> t1, Tuple2<Long, Long> accumulator) {
        return new Tuple2<>(accumulator.f0 + t1.f1, accumulator.f1 + 1L);
    }


    @Override
    public Tuple2<Long, Long> merge(Tuple2<Long, Long> acc1, Tuple2<Long, Long> acc2) {
        return new Tuple2<>(acc1.f0 + acc2.f0, acc1.f1 + acc2.f1);
    }

    @Override
    public Double getResult(Tuple2<Long, Long> accumulator) {
        return ((double)accumulator.f0 / accumulator.f1);
    }

}
